
import java.util.UUID

import commons.conf.ConfigurationManager
import commons.constant.Constants
import commons.utils.ParamUtils
import net.sf.json.JSONObject
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SaveMode, SparkSession}


/**
 * 需求六：各区域Top3商品统计
 */
object AreaTop3Stat {




  def main(args: Array[String]): Unit = {
    val jsonStr: String = ConfigurationManager.config.getString(Constants.TASK_PARAMS)
    val taskParam: JSONObject = JSONObject.fromObject(jsonStr)

    val taskUUID: String = UUID.randomUUID().toString

    val conf: SparkConf = new SparkConf().setAppName("areaTop3Stat").setMaster("local[*]")

    val sparkSession: SparkSession = SparkSession.builder().config(conf).enableHiveSupport().getOrCreate()

    //  获取（城市id，点击商品的id）=》(4,1)
    val cityId2PidRDD = getCityAndProductInfo(sparkSession, taskParam)
    //  创建城市区域信息
    val cityId2AreaInfoRDD: RDD[(Long, CityAreaInfo)] = getCityAreaInfo(sparkSession)
    // 创建一个临时表保存（"city_id","city_name","area","pid"），代表一次的点击商品行为
    getAreaPidBasicInfoTable(sparkSession, cityId2PidRDD, cityId2AreaInfoRDD)

    sparkSession.udf.register("concat_long_string", (v1: Long, v2: String, split: String) => {
      v1 + split + v2
    })

    sparkSession.udf.register("group_concat_distinct", new GroupConcatDistinct)

    // 创建一个临时表，按照area，pid分组统计点击次数
    getAreaProductCilckCountTable(sparkSession)

    sparkSession.udf.register("get_json_field", (json:String, field:String) => {
      val jsonObject = JSONObject.fromObject(json)
      jsonObject.getString(field)
    })

    getAreaProductClickCountInfo(sparkSession)
    getTop3Product(sparkSession, taskUUID)
    sparkSession.sql("select * from tmp_area_count_product_info").show()

  }
  def getTop3Product(sparkSession: SparkSession, taskUUID: String) = {
    /*    val sql = "select area, city_infos, pid, product_name, product_status, click_count, " +
          "row_number() over(PARTITION BY area ORDER BY click_count DESC) rank from tmp_area_count_product_info"

        sparkSession.sql(sql).createOrReplaceTempView("temp_test")*/

    val sql = "select area, " +
      "CASE " +
      "WHEN area='华北' OR area='华东' THEN 'A_Level' " +
      "WHEN area='华中' OR area='华南' THEN 'B_Level' " +
      "WHEN area='西南' OR area='西北' THEN 'C_Level' " +
      "ELSE 'D_Level' " +
      "END area_level, " +
      "city_infos, pid, product_name, product_status, click_count from (" +
      "select area, city_infos, pid, product_name, product_status, click_count, " +
      "row_number() over(PARTITION BY area ORDER BY click_count DESC) rank from " +
      "tmp_area_count_product_info) t where rank<=3"

    val top3ProductRDD = sparkSession.sql(sql).rdd.map{
      case row =>
        AreaTop3Product(taskUUID, row.getAs[String]("area"), row.getAs[String]("area_level"),
          row.getAs[Long]("pid"), row.getAs[String]("city_infos"),
          row.getAs[Long]("click_count"), row.getAs[String]("product_name"),
          row.getAs[String]("product_status"))
    }

    import sparkSession.implicits._
    top3ProductRDD.toDF().write
      .format("jdbc")
      .option("url", ConfigurationManager.config.getString(Constants.JDBC_URL))
      .option("dbtable", "area_top3_product_0308")
      .option("user", ConfigurationManager.config.getString(Constants.JDBC_USER))
      .option("password", ConfigurationManager.config.getString(Constants.JDBC_PASSWORD))
      .mode(SaveMode.Append)
      .save()
  }
  /**
   *
   * @param sparkSession
   */
  def getAreaProductClickCountInfo(sparkSession: SparkSession) = {
    val sql = "select tacc.area, tacc.city_infos, tacc.pid, pi.product_name, " +
      "if(get_json_field(pi.extend_info, 'product_status')='0','Self','Third Party') product_status," +
      "tacc.click_count " +
      " from tmp_area_click_count tacc join product_info pi on tacc.pid = pi.product_id"

    sparkSession.sql(sql).createOrReplaceTempView("tmp_area_count_product_info")
  }
  /**
   *
   * @param sparkSession
   */
  def getAreaProductCilckCountTable(sparkSession: SparkSession): Unit = {
    val sql = "select area,pid,count(*) click_count ,group_concat_distinct(concat_long_string(city_id,city_name,':')) city_infos from tmp_area_basic_info group by area,pid"
    sparkSession.sql(sql).createTempView("tmp_area_click_count")
  }

  /**
   * 创建一个临时表保存（"city_id","city_name","area","pid"），代表一次的点击商品行为
   *
   * @param sparkSession
   * @param cityId2PidRDD
   * @param cityId2AreaInfoRDD
   */
  def getAreaPidBasicInfoTable(sparkSession: SparkSession,
                               cityId2PidRDD: RDD[(Long, Long)],
                               cityId2AreaInfoRDD: RDD[(Long, CityAreaInfo)]) = {

    val areaPidInfoRDD: RDD[(Long, String, String, Long)] = cityId2PidRDD.join(cityId2AreaInfoRDD).map {
      case (cityId, (pid, areaInfo)) =>
        (cityId, areaInfo.city_name, areaInfo.area, pid)
    }

    import sparkSession.implicits._
    areaPidInfoRDD.toDF("city_id", "city_name", "area", "pid").createTempView("tmp_area_basic_info  ")

  }

  /**
   * 获取城市区域信息
   *
   * @param sparkSession
   */
  def getCityAreaInfo(sparkSession: SparkSession) = {
    val cityAreaInfoArray = Array((0L, "北京", "华北"), (1L, "上海", "华东"), (2L, "南京", "华东"),
      (3L, "广州", "华南"), (4L, "三亚", "华南"), (5L, "武汉", "华中"),
      (6L, "长沙", "华中"), (7L, "西安", "西北"), (8L, "成都", "西南"),
      (9L, "哈尔滨", "东北"))

    sparkSession.sparkContext.makeRDD(cityAreaInfoArray).map {
      case (cityId, cityName, cityArea) =>
        (cityId, CityAreaInfo(cityId, cityName, cityArea))
    }

  }


  def getCityAndProductInfo(sparkSession: SparkSession, taskParam: JSONObject): RDD[(Long, Long)] = {

    val startDate: String = ParamUtils.getParam(taskParam, Constants.PARAM_START_DATE)
    val endDate: String = taskParam.getString(Constants.PARAM_END_DATE)


    val sql = "select city_id,click_product_id from user_visit_action where date >='" + startDate + "'and date <='" + endDate + "'and click_product_id != -1"

    import sparkSession.implicits._
    sparkSession.sql(sql).as[CityClickProduct].rdd.map {
      case cityPid => (cityPid.city_id, cityPid.click_product_id)
    }
  }
}
